package org.ourgrid.broker.business.requester;

import java.io.IOException;
import java.util.ArrayList;
import java.util.List;

import org.ourgrid.broker.BrokerConstants;
import org.ourgrid.broker.business.dao.BrokerDAOFactory;
import org.ourgrid.broker.business.dao.BrokerMessageProcessorDAO;
import org.ourgrid.broker.business.messages.BrokerControlMessages;
import org.ourgrid.broker.business.scheduler.SchedulerIF;
import org.ourgrid.broker.business.scheduler.workqueue.WorkQueueReplication;
import org.ourgrid.broker.business.schedulerdata.SchedulerData;
import org.ourgrid.broker.business.schedulerdata.executor.control.SchedulerDataExecutorControl;
import org.ourgrid.broker.communication.actions.SchedulerAction;
import org.ourgrid.broker.communication.processors.ErrorOcurredProcessor;
import org.ourgrid.broker.communication.processors.HereIsFileInfoProcessor;
import org.ourgrid.broker.communication.processors.HereIsGridProcessResultProcessor;
import org.ourgrid.broker.communication.processors.HereIsWorkerSpecProcessor;
import org.ourgrid.broker.communication.processors.WorkerIsReadyProcessor;
import org.ourgrid.broker.communication.receiver.LocalWorkerProviderClientReceiver;
import org.ourgrid.broker.communication.receiver.WorkerClientReceiver;
import org.ourgrid.broker.request.StartBrokerRequestTO;
import org.ourgrid.common.internal.IResponseTO;
import org.ourgrid.common.internal.RequesterIF;
import org.ourgrid.common.internal.response.CreateRepeatedActionResponseTO;
import org.ourgrid.common.internal.response.DeployServiceResponseTO;
import org.ourgrid.common.internal.response.LoggerResponseTO;
import org.ourgrid.common.job.JobCounter;
import org.ourgrid.common.job.PersistentJobCounter;
import org.ourgrid.common.job.SimpleJobCounter;

public class StartBrokerRequester implements RequesterIF<StartBrokerRequestTO> {

	public List<IResponseTO> execute(StartBrokerRequestTO request) {
		BrokerDAOFactory.getInstance().reset();
		
		List<IResponseTO> responses = new ArrayList<IResponseTO>();
		
		createServices(responses);
		createRepetedActions(responses);
		
		JobCounter jobCounter = createJobCounter(request);
		
		BrokerDAOFactory.getInstance().getJobCounterDAO().setJobCounter(jobCounter);
		
		initSchedulers(request);
		
		List<SchedulerData> outputCommands = new ArrayList<SchedulerData>();
		for(SchedulerIF scheduler: BrokerDAOFactory.getInstance().getJobDAO().getSchedulers()) {
			List<SchedulerData> cjData = scheduler.start();
			if (cjData != null) {
				outputCommands.addAll(cjData);
			}	
		}
		
		responses.addAll(SchedulerDataExecutorControl.getInstance().execute(outputCommands));
		
		responses.add(new LoggerResponseTO(BrokerControlMessages.getSuccessfullyStartedBrokerMessage(), LoggerResponseTO.INFO));
		
		
		return responses;
	}
	
	protected void createServices(List<IResponseTO> responses) {
		DeployServiceResponseTO deployTO = new DeployServiceResponseTO();
		deployTO.setServiceName(BrokerConstants.LOCAL_WORKER_PROVIDER_CLIENT);
		deployTO.setServiceClass(LocalWorkerProviderClientReceiver.class);
		responses.add(deployTO);
		
		deployTO = new DeployServiceResponseTO();
		deployTO.setServiceName(BrokerConstants.WORKER_CLIENT);
		deployTO.setServiceClass(WorkerClientReceiver.class);
		responses.add(deployTO);

		createMessageProcessors();
	}
	
	private void createMessageProcessors() {
		BrokerMessageProcessorDAO brokerMessageProcessorDAO = BrokerDAOFactory.getInstance().getBrokerMessageProcessorDAO();
		
		brokerMessageProcessorDAO.
			putMessageProcessor(BrokerConstants.WORKER_IS_READY_ACTION_NAME, new WorkerIsReadyProcessor());
		brokerMessageProcessorDAO.
			putMessageProcessor(BrokerConstants.HERE_IS_EXECUTION_RESULT_ACTION_NAME, new HereIsGridProcessResultProcessor());
		brokerMessageProcessorDAO.
			putMessageProcessor(BrokerConstants.ERROR_OCURRED_ACTION_NAME, new ErrorOcurredProcessor());
		brokerMessageProcessorDAO.
			putMessageProcessor(BrokerConstants.HERE_IS_FILE_INFO_ACTION_NAME, new HereIsFileInfoProcessor());
		brokerMessageProcessorDAO.
			putMessageProcessor(BrokerConstants.HERE_IS_WORKER_SPEC_ACTION_NAME, new HereIsWorkerSpecProcessor());
	}
	
	private void initSchedulers(StartBrokerRequestTO request) {
		
		String maxReplicas = request.getMaxReplicas();
		String maxFails = request.getMaxFails();
		String maxBlFails = request.getMaxBlackListFails();
		
		SchedulerIF scheduler = new WorkQueueReplication(Integer.valueOf(maxReplicas), 
				Integer.valueOf(maxFails), Integer.valueOf(maxBlFails));
		
		BrokerDAOFactory.getInstance().getJobDAO().addScheduler(scheduler);
	}

	private void createRepetedActions(List<IResponseTO> responses) {
		CreateRepeatedActionResponseTO to = new CreateRepeatedActionResponseTO();
		to.setActionName(BrokerConstants.SCHEDULER_ACTION_NAME);
		to.setRepeatedAction(new SchedulerAction());
		
		responses.add(to);
	}
	
	/**
	 * Factory method for the JobCounter.
	 * @return JobCounter based on the Broker Context
	 */
	private JobCounter createJobCounter(StartBrokerRequestTO request) {
		
		if (request.isPersistentJobEnable()) {
			try {
				return new PersistentJobCounter(request.getJobCounterFilePath());
			} catch (IOException e) { }
		}
		return new SimpleJobCounter();
	}
}
